package kk

import (
	"context"
	"errors"
	"github.com/segmentio/kafka-go"
	"log"
	"time"
)

// KafkaWriter （生产者）消息写入者
type KafkaWriter struct {
	w    *kafka.Writer
	data chan LogData
}

// GetWriter kafka生产消息
func GetWriter(addr string) *KafkaWriter {
	// make a writer that produces to topic-A, using the least-bytes distribution
	w := &kafka.Writer{
		//设置kafka服务端口  一个Kafka生产者只需连接到其中的一个Broker（kafka服务器）
		//因为Kafka集群内的所有Broker都会自动进行数据复制和分布
		Addr: kafka.TCP(addr),
		//设置负载均衡选择 partition
		Balancer: &kafka.LeastBytes{},
	}
	k := &KafkaWriter{
		w:    w,
		data: make(chan LogData, 100),
	}
	//写入消息到kafka
	go k.sendKafka()
	return k
}

type LogData struct {
	Topic string
	Data  []byte
}

func (w *KafkaWriter) Send(data LogData) {
	w.data <- data

}
func (w *KafkaWriter) Close() {
	if w.w != nil {
		w.w.Close()
	}
}

// 写入logMsg数据到kafka
func (w *KafkaWriter) sendKafka() {
	for {
		select {
		case data := <-w.data:
			//准备好需要写入的数据
			messages := []kafka.Message{
				{
					Topic: data.Topic,
					Key:   []byte("logMsg"),
					Value: data.Data,
				},
			}

			var err error
			const retries = 3
			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
			defer cancel()
			//for循环3次目的：第一次写入没有topic会写入失败，所以第一次写入就是为了创建topic  然后再写入 ，
			for i := 0; i < retries; i++ {

				// 批量写入数据
				err = w.w.WriteMessages(ctx, messages...)
				if err == nil {
					break
				}
				//这里休息，是因为第一次写入 去创建topic去了
				if errors.Is(err, kafka.LeaderNotAvailable) || errors.Is(err, context.DeadlineExceeded) {
					time.Sleep(time.Millisecond * 250)
					continue
				}
				//其他错误打印出来看看
				if err != nil {
					log.Printf("kafka send writemessage err %s", err.Error())
				}
			}
		}
	}

}
